package Schedule::Chronic::Distributed::Task;

=head1 NAME

  Schedule::Chronic::Distributed::Task - Class representing Task within Schedule::Chronic::Distributed

=head1 SYNOPSIS

  use Schedule::Chronic::Distributed::Task;

=head1 DESCRIPTION

This class represents Tasks within the schedule, i.e. jobs that need doing, holding their state, arguments and
related date and historical data.

This class provides access to related data/classes such as the command, node, and a history of when the job
was run, due, and it's previous states and errors.

=head1 METHODS

=head2 ACCESSORS

=over

=item id 

=item name

=item command

=item command_arguments 

=item priority

=item created timeout

=item min_interval

=item max_interval

=item retry_interval

=item schedule_type

=item notify_address

=item current_status

=item current_node

=item current_due

=item current_deadline

=item paused

=item last_status

=item last_due

=item last_deadline

=item last_run

=item last_completed

=item last_node

=item description

=item error

=item warning

=item category

=back

=cut

use strict;

use Time::Piece;
use Time::Seconds;

use base qw(Schedule::Chronic::Distributed::DBI);

use Schedule::Chronic::Distributed::TaskHistory;

__PACKAGE__->table('task');
__PACKAGE__->sequence('task_id_seq');
__PACKAGE__->columns(All => qw/id name command command_arguments priority created timeout
		     min_interval max_interval retry_interval schedule_type notify_address
		     current_status current_node current_due current_deadline paused
		     last_status last_due last_deadline last_run last_completed last_node
		     description error warning category/);
__PACKAGE__->columns(Primary => 'id');
__PACKAGE__->columns(TEMP => qw/constraint_objects wait/);

__PACKAGE__->has_many(task_times => 'Schedule::Chronic::Distributed::TaskTimes');

__PACKAGE__->has_a(current_node => 'Schedule::Chronic::Distributed::Node');
__PACKAGE__->has_a(last_node => 'Schedule::Chronic::Distributed::Node');
__PACKAGE__->has_a(command => 'Schedule::Chronic::Distributed::Command');
__PACKAGE__->has_a(category => 'Schedule::Chronic::Distributed::Category');

__PACKAGE__->has_many(prerequisites => 'Schedule::Chronic::Distributed::PrerequisiteTask');


#
# Date fields
__PACKAGE__->has_a( created => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'created'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );
# Previous dates
__PACKAGE__->has_a( last_run => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'last_run'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );
__PACKAGE__->has_a( last_completed => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'last_completed'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );
__PACKAGE__->has_a( last_due => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'last_due'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );
__PACKAGE__->has_a( last_deadline => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'last_deadline'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );
# Current dates
__PACKAGE__->has_a( current_deadline => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'current_deadline'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );

__PACKAGE__->has_a( current_due => 'Time::Piece',
		    inflate => sub { return __PACKAGE__->inflate_timepiece(@_,'current_due'); },
		    deflate => sub { return __PACKAGE__->deflate_timepiece(@_) } );

#
# set defaults

__PACKAGE__->add_trigger(before_create => sub {
   my $now = localtime;
   $_[0]->set(created => $now->strftime("%Y-%m-%d %T"));
   $_[0]->set(current_status => 'new');
});

__PACKAGE__->add_trigger(after_create => sub {
    my $self = shift;
    unless ($self->current_due or $self->schedule_type =~ 'times') {
      $self->_update_times;
    }
    unless (defined $self->paused) { $self->paused(0); $self->update(); }
});


#
# Set constraints
__PACKAGE__->constrain_column(schedule_type => [qw/times times_repeat intervals prerequisite/]);
__PACKAGE__->constrain_column(current_status => ['awaiting prerequisite',qw/complete error idle new running/]);

########################################
# Custom Queries

__PACKAGE__->set_sql ( first_tasktime => <<END_NEXT_TASKTIME_SQL, 0);
  SELECT  MIN(start_time) AS start_time,
  MIN(end_time) AS end_time
  FROM task_times tt
  WHERE tt.task = ?
END_NEXT_TASKTIME_SQL

__PACKAGE__->set_sql ( waiting_tasks => <<END_WAITING_SQL, 0);
SELECT task, requisite_state, latency
FROM prerequisite_tasks, task
WHERE requisite = ?
AND task.id = prerequisite_tasks.task
AND task.current_status = 'awaiting prerequisite'
END_WAITING_SQL

__PACKAGE__->set_sql ( requisite_tasks => <<END_REQUISITE_SQL, 0);
SELECT MIN(latency) as latency
FROM prerequisite_tasks, task as required_task, task as dependant_task
WHERE dependant_task.id = prerequisite_tasks.task
AND dependant_task.id = ?
AND required_task.id = prerequisite_tasks.requisite
AND dependant_task.last_run < required_task.last_run
AND required_task.current_status = prerequisite_tasks.requisite_state
END_REQUISITE_SQL

__PACKAGE__->set_sql ( next_tasks => <<END_NEXT_SQL, 0);
SELECT 
distinct on (substring(to_char(current_deadline,'YYYYMMDDHHSS') from 1 for 11),
priority,
substring(to_char(current_due,'YYYYMMDDHHSS') from 1 for 11),
t.id) t.*
FROM
task AS t,
(
  SELECT task.id as task_id, 
  MIN ( %s ) AS constraints_met
  FROM task left outer join task_requirement tr on task.id = tr.task
  GROUP BY task.id
) AS _trs
WHERE
_trs.task_id = t.id
AND (
 ( (_trs.constraints_met = 1 or _trs.constraints_met is null) AND current_due < now() )
 OR ( current_deadline < now() )
)
AND (paused = 0 OR paused is NULL)
ORDER BY
substring(to_char(current_deadline,'YYYYMMDDHHSS') from 1 for 11),
priority,
substring(to_char(current_due,'YYYYMMDDHHSS') from 1 for 11),
t.id
LIMIT
%s
END_NEXT_SQL

__PACKAGE__->set_sql ( next_tasktime => <<END_NEXT_TASKTIME_SQL, 0);
  SELECT MIN(start_time) AS start_time,
  MIN(end_time) AS end_time
  FROM task_times tt
  WHERE tt.task = ?
  AND %s > '%s'
END_NEXT_TASKTIME_SQL

__PACKAGE__->set_sql ( catch_orphaned_running_tasks => <<END_ORPHAN_TASKS_SQL, 0);
  UPDATE task
  SET current_status = 'error', error='node timed out or stopped before completing this task'
  FROM nodes
  WHERE current_status = 'running' and task.current_node = nodes.id and nodes.available = 0
END_ORPHAN_TASKS_SQL

########################################
# Accessors and Attributes

=head2 Class Methods

Class::DBI methods provided are provided by this class and include..

=over

=item retrieve

fetch a single object from id

=item retrieve_all

fetch all objects of this class

=item search

search for and fetch all objects of this class based on given criteria

=item create

create and insert a new object into the database

=back

Other class methods are..

=over

=item search_current

fetch objects based on deadline

=item get_next_tasks

fetch next object due, based on deadline, due time and constraints

=cut

sub get_next_tasks {
  my ($class,$node,$count) = @_;
  warn "get_next_tasks\n";
  $count ||= 10;
  my $constraint_conditions = $node->constraints_as_sql;
  my $sth = $class->sql_next_tasks($constraint_conditions, $count );
#  warn "sth args : $constraint_conditions, $count\n";
  my @tasks = $class->sth_to_objects($sth);
  $sth->finish();
  $class->db_Main->{CachedKids} = {};
  return @tasks;
}

=item check_running_jobs

check the database for tasks running nodes that have gone away, or have
failed to timeout on the given node.

=item catch_orphaned_jobs

Checks for and updates tasks in the database that have been orphaned by
nodes that were running but are now absent.

=cut

sub catch_orphaned_jobs {
    my $class = shift;
    my $sth = $class->sql_catch_orphaned_running_tasks();
    my $rv = $sth->execute();
}

=back

=head2 METHODS PROVIDED ELSEWHERE

Several additional methods for the Monitor Web UI are provided in
Schedule::Chronic::Distributed::Monitor::Task package.

These are :

=over

=item add_time

=item running

=item overdue

=item errors

=item warnings

=item pause

=item continue

=back

=cut

########################################
# Public object action methods

=head1 Object Action Methods

=head2 reserve

  Attempts to reserve an action for a node to run, using transactions and locking
  sets current_node and and current_status within a lock

  my $reserved = $task->reserve();

=cut

sub reserve {
  my ($self,$node) = @_;
  my $id = $self->id;
#  warn "reserving task for node : $node\n";
  my $dbh = $self->db_Main;
  my $reserved = 0;
  # Force serialisation of object/database action within this block
  RESERVATION_TRANSACTION : {
    $self->remove_from_object_index();
    my $rc = $dbh->begin_work or die $dbh->errstr;
    my $locked = $dbh->do('LOCK TABLE task IN ROW EXCLUSIVE MODE');
    # make sure our new version of $self has latest values from table
    # (which is now locked)
    $self = __PACKAGE__->retrieve($id);

    if ($self->current_status eq 'running') {
      warn "can't reserve - currently running\n";
      $dbh->rollback;
      return 0;
    }

    $self->current_node($node->id);
    $self->current_status('running');
    $self->error('');
    $self->warning('');
    my $t = localtime;
    $self->last_run($t);
    $self->update;
    $reserved = 1;
#    my $unlocked = $dbh->do('UNLOCK TABLES'); # MySQL
    $dbh->commit;

  } # end RESERVATION_TRANSACTION
#  warn "reserved ok : $reserved\n";
  return $reserved;
}

=head2 run

  Attempts to run the command for the task, updating times, status, etc.

  Returns true for success, undefined/false for failure.

  Possibility of failing to return if uncatchable error occurs, or if fatal
  error occurs that makes continuing likely to corrupt data or otherwise do
  harm to the system or application data.

  Requires a Node object, and optionally a path to scripts

  my $completed = $task->run($NodeObject,'/path/to/scripts');

=cut

sub run {
  my ($self,$node,$path) = @_;

  # Ensure only 1 node can run a task at the same time
  unless ($self->reserve($node)) {
    warn "task already running, couldn't reserve, giving up\n";
    return 0;
  }

  # get path subdirectory from category
  if ($path && $self->category->path) {
    $path .= '/'.$self->category->path . '/';
  }

  # The task in the database has been updated and is running exclusively
  # on this node
  my $command = $self->command;
  my ($ran_ok,$error) = $command->run($self,$path);
  warn "command ran ok ? : $ran_ok\n";
  if ($ran_ok) {
    $self->current_status('complete');
    my $t = localtime;
    $self->last_completed($t);

    # check for warnings, and use them if present
    if ($ran_ok) {
      my $warning_filename = "/tmp/schedule_task_".$$.".warning";
      if (-f $warning_filename && -r $warning_filename) {
	open(WARNING,"<$warning_filename") or warn "couldn't open $warning_filename to capture warning for pid $$ : $!";
	$self->warning( join('',<WARNING>) );
	close(WARNING);
      }
    }
  } else {
    warn "Error running task : ", $self->id, " pid : $$ details : $error\n";
    $error ||= 'N/A - sorry';
    $self->current_status('error');
    # FIXME: provide decent error message, pull from stderr/stdout file
    # as with timeouts

    my $stdout_filename = "/tmp/schedule_task_$$".".stdout";
    my $stderr_filename = "/tmp/schedule_task_$$".".stderr";

    open(CAPTURED_STDERR,"<$stderr_filename") or warn "couldn't open $stderr_filename to capture STDERR for pid $$ : $!";
    my $stderr = join('',<CAPTURED_STDERR>);
    close(CAPTURED_STDERR);

    open(CAPTURED_STDOUT,"<$stdout_filename") or warn "couldn't open $stdout_filename to capture STDOUT for pid $$ : $!";
    my $stdout = join('',<CAPTURED_STDOUT>);
    close(CAPTURED_STDOUT);

    my $error_report = "Process $$ $error \n\nSTDERR:\n$stderr\n\nSTDOUT:\n$stdout";

    $self->error($error_report);
  }
  $self->update();

  $self->_update_waiting_tasks();

  $self->notify();
  # update historical values
  $self->_update_history();

  # update current/next times
  $self->_update_times();

  $self->update();
  return $ran_ok;
}

=head2 notify

 sends a notification email to the address in notify_address

=cut

sub notify {
  my $self = shift;
  return unless ($self->current_status eq 'error');
  # FIXME: use configuration to get type of mailer and mailer args
  # Notify the email address.
  if (my $to_address = $self->notify_address) {
    my $from_address = 'scheduler@localhost';
    my $taskname = $self->name;
    my $error = $self->error;
    my $node = $self->last_node;
    my $time = localtime;
    my $message = <<END_NOTIFY_MESSAGE;
To: $to_address
From: $from_address
Subject: Notification of Failure in scheduled task : $taskname

Error message for task $taskname at $time on node $node follows:

$error

--
Generated by automated scheduler on $node

END_NOTIFY_MESSAGE
    my $sender = Email::Send->new({mailer => 'Sendmail'});
    $sender->send($message);
  }
}

########################################
# Internal object action methods

sub _update_history {
  my $self = shift;
  # move current values to last values
  $self->last_status($self->current_status());
  $self->last_due ($self->current_due());
  $self->last_deadline($self->current_deadline());
  $self->last_node($self->current_node());
  $self->update();

  # populate history from last values
  my $history = Schedule::Chronic::Distributed::TaskHistory->create_from_task($self);
#  warn "added to history : $history\n";
  $self->update();

  return;
}


sub _update_times {
  my $self = shift;
  # set current due, overdue etc from task_times
  my $dbh = $self->db_Main();

  # check status
  my $status = $self->current_status;
  my ($due, $deadline);
  my $now = localtime;
  if ($status ne 'complete') {
    if ($self->retry_interval) {
      $due = $now + $self->retry_interval;
      $deadline = $due + 120;
    } elsif ( $self->min_interval) {
      $due = $now + $self->min_interval;
      $deadline = $now + $self->max_interval;
    }  else {
      $due = $now + 240; # 3 mins
      $deadline = $due + 180; # 6 mins
    }
  } else {
  SCHEDULE_TYPE: {
      if ($self->schedule_type =~ /times/i) {
	my $next_time_field = (lc($self->schedule_type) eq 'times_repeat') ? 'end_time' : 'start_time';
	my $sth = $self->sql_next_tasktime($next_time_field, $now->time);
	my $rv = $sth->execute($self->id);
	my ($due_time, $deadline_time) = $sth->fetchrow_array();
	$sth->finish();

	unless ($due_time && $deadline_time ) {
	  $sth = $self->sql_first_tasktime();
	  my $rv = $sth->execute($self->id);
	  ($due_time, $deadline_time) = $sth->fetchrow_array();
	  unless ($due_time && $deadline_time) {
	    warn "couldn't get times for task ", $self->id;
	  }
	  $now += ONE_DAY;
	  $sth->finish();
	}

	unless ($deadline_time && $due_time) {
	  warn "missing deadline_time and due_time for task ", $self->id, "\n";
	  return 0;
	}

	my $day = $now->ymd;
	$due = Time::Piece->strptime("$day $due_time", "%Y-%m-%d %H:%M:%S");
	$deadline = Time::Piece->strptime("$day $deadline_time", "%Y-%m-%d %H:%M:%S");

	last SCHEDULE_TYPE;
      }

      if ($self->schedule_type eq 'prerequisite') {
	# check if a pre-requisite has completed and it's time to run again
	my $sth = $self->sql_requisite_tasks();
	my $rv = $sth->execute($self->id);
	my ($pre_requisite_latency) = $sth->fetchrow_array();
	$sth->finish();
	if ($pre_requisite_latency) {
	  $due = $now;
	  $deadline = $now + $pre_requisite_latency;
	  $self->current_status('idle');
	} else {
	  $self->current_status('awaiting prerequisite');
	}
	last SCHEDULE_TYPE;
      }
      if ($self->schedule_type eq 'intervals') {
	if ($self->min_interval) {
	  $due = $now + $self->min_interval;
	  $deadline = $now + $self->max_interval;
	} else {
	  warn "no min interval for this task : ", $self->id;
	}
      }
    }				# end SCHEDULE_TYPE;
  }
  $self->current_due($due);
  $self->current_deadline($deadline);
  $self->_update_waiting_tasks();
  $self->update();

  return;
}

sub _update_waiting_tasks {
  my $self = shift;
  warn "updating waiting tasks..\n";
  my $sth = $self->sql_waiting_tasks();
  my $rv = $sth->execute($self->id);
  my $status = $self->current_status;
  while (my $waiting_task = $sth->fetchrow_hashref) {
    warn "waiting task : ", join ( ', ', %$waiting_task), "\n";
    if ($waiting_task->{requisite_state} eq $status) {
      my $task = __PACKAGE__->retrieve($waiting_task->{task});
      unless ($task) {
	warn "non-existant task ", $waiting_task->{task}, "has task ", $self->id, " as requisite";
	next;
      }
      warn "updating task : ", $task->id, " : ", $task->name, "\n";
      my $t = localtime;
      $task->current_due($t);
      $task->current_deadline($t);
      $task->current_status('idle');
      $task->update;
    }
  }
  $sth->finish();
}

###############################################################################

=head1 SEE ALSO

Schedule::Chronic::Distributed

Schedule::Chronic::Distributed::Node

Schedule::Chronic::Distributed::Monitor::Task

Schedule::Chronic::Distributed::Command

Class::DBI

Schedule::Chronic

=head1 AUTHOR

aaron trevena, E<lt>teejay@droogs.orgE<gt>

=head1 COPYRIGHT AND LICENSE

Copyright (C) 2006 Aaron Trevena, Aviation Briefing Ltd.

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.8.5 or,
at your option, any later version of Perl 5 you may have available.


=cut

1;
